package com.csw.flink.transformation

import com.csw.flink.source.Demo03MysqlSource
import com.csw.flink.source.Demo03MysqlSource.MysqlSource
import org.apache.flink.streaming.api.scala._

object Demo05Agg {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val scoreDS: DataStream[Demo03MysqlSource.Score] = env.addSource(new MysqlSource)


    val keyByDS: KeyedStream[Demo03MysqlSource.Score, Int] = scoreDS.keyBy(_.studentId)

    /**
      *
      * sum
      * max
      * maxBy
      * min
      * minBy
      */

    /**
      * max和min返回流中的最大值和最小值，但是maxBy和minBy返回最大值以及最小值所对应的所有元素。
      */

      //返回最小值所对应的所有元素
    val ds1: DataStream[Demo03MysqlSource.Score] = keyByDS.minBy("score")

    //返回最小值
    val ds2: DataStream[Demo03MysqlSource.Score] = keyByDS.min("score")


    ds2.print()

    env.execute()
  }
}
